#include "../Include/ThreadPool2.h"

CThreadPool2::CThreadPool2()
	: m_mutex(),
	m_condNotEmpty(m_mutex),
	m_condNotFull(m_mutex),
	m_nMaxQueueSize(0),
	m_bRunning(False)
{

}

CThreadPool2::~CThreadPool2()
{
	if (m_bRunning)
	{
		Stop();
	}
}

void CThreadPool2::Start(Int32 nNumThreads, Int32 nCoreIdx)
{
	m_bRunning = true;
	m_vecThreads.reserve(nNumThreads);
	for (Int32 i = 0; i < nNumThreads; ++i)
	{
		CThread *pThread = new CWorkerThread(nCoreIdx, this);
		pThread->Start();
		m_vecThreads.push_back(pThread);
	}
}

void CThreadPool2::Stop()
{
	{
		CMutexLocker lock(&m_mutex);
		m_bRunning = False;
		m_condNotEmpty.NotifyAll();
	}
	
	vector<CThread*>::iterator it = m_vecThreads.begin();
	while (it != m_vecThreads.end())
	{
		(*it)->Stop();
		it++;
	}
}

UInt32 CThreadPool2::GetQueueSize() const
{
	CMutexLocker lock(&m_mutex);
	return m_queueTask.size();
}

void CThreadPool2::Run(TaskFunc pFunc, void *pArg)
{
	if (m_vecThreads.empty())
	{
		pFunc(pArg);
	}
	else
	{
		CMutexLocker lock(&m_mutex);
		while (IsFull())
		{
			m_condNotFull.Wait();
		}
		
		Task* pTask = new Task;
		pTask->pFunc = pFunc;
		pTask->pArg = pArg;
		m_queueTask.push_back(pTask);
		m_condNotEmpty.Notify();
	}
}


Task* CThreadPool2::Take()
{
	CMutexLocker lock(&m_mutex);

	while (m_queueTask.empty() && m_bRunning)
	{
		m_condNotEmpty.Wait();
	}

	Task* task;
	if (!m_queueTask.empty())
	{
		task = m_queueTask.front();
		m_queueTask.pop_front();
		if (m_nMaxQueueSize > 0)
		{
			m_condNotFull.Notify();
		}
	}

	return task;
}

Bool CThreadPool2::IsFull() const
{
	return m_nMaxQueueSize > 0 && m_queueTask.size() >= m_nMaxQueueSize;
}

void CThreadPool2::RunInThread()
{
	while (m_bRunning)
	{
		Task* pTask = Take();
		if (pTask->pFunc)
		{
			pTask->pFunc(pTask->pArg);
			delete pTask;
		}
	}
}

CWorkerThread::CWorkerThread(Int32 nCPUIdx, CThreadPool2* pThreadPool)
	: m_nCPUIndex(nCPUIdx),
	m_pThreadPool(pThreadPool)
{
}

CWorkerThread::~CWorkerThread()
{
	Stop();
}

Bool CWorkerThread::OnCreateThread(UInt32 &dwStackSize, Int32 &nCPUIndex)
{
	nCPUIndex = m_nCPUIndex;
	return CThread::OnCreateThread(dwStackSize, nCPUIndex);
}

void CWorkerThread::OnRun()
{
	if (m_pThreadPool)
	{
		m_pThreadPool->RunInThread();
	}
}


